package com.arch.flink.archetype.bulk;

import org.apache.flink.api.common.serialization.BulkWriter;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.connector.file.table.factories.BulkWriterFormatFactory;
import org.apache.flink.core.fs.FSDataOutputStream;
import org.apache.flink.core.fs.FileSystem;
import org.apache.flink.core.fs.Path;
import org.apache.flink.table.connector.format.EncodingFormat;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.factories.FactoryUtil;
import org.apache.flink.table.types.DataType;
import org.apache.flink.table.types.logical.RowType;
import org.apache.flink.table.types.utils.LogicalTypeDataTypeConverter;

import java.io.IOException;

import static org.apache.flink.table.types.utils.TypeConversions.fromLogicalToDataType;

/**
 * @author pizhihui
 * @date 2023-08-13
 */
public class BulkWriterDemo {

    public static void main(String[] args) throws IOException {


        String tmpDir = System.getProperty("java.io.tmpdir");
        System.out.println("tmpDir: " + tmpDir);

        BulkWriterFormatFactory fav = FactoryUtil.discoverFactory(BulkWriterDemo.class.getClassLoader(), BulkWriterFormatFactory.class, "fav");

        Configuration conf = new Configuration();

        RowType type = new RowType(null);
        DataType dataType = LogicalTypeDataTypeConverter.toDataType(type);

        EncodingFormat<BulkWriter.Factory<RowData>> encodingFormat = fav.createEncodingFormat(null, conf);
        BulkWriter.Factory<RowData> runtimeEncoder = encodingFormat.createRuntimeEncoder(null, dataType);


        Path path = new org.apache.flink.core.fs.Path(tmpDir, "");
        FileSystem fileSystem = path.getFileSystem();

        FSDataOutputStream out = fileSystem.create(path, FileSystem.WriteMode.OVERWRITE);

        BulkWriter<RowData> writer = runtimeEncoder.create(out);

        RowData t1 = GenericRowData.of(1, 2);
        writer.addElement(t1);

        writer.flush();
        writer.finish();

        out.close();

    }

}
